package org.apache.rocketmq.broker.util;

import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.config.BrokerRole;

import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;

public class HookUtils {

    protected static final Logger LOG = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);

    private static final AtomicLong PRINT_TIMES = new AtomicLong(0);

    /**
     * On Linux: The maximum length for a file name is 255 bytes.
     * The maximum combined length of both the file name and path name is 4096 bytes.
     * This length matches the PATH_MAX that is supported by the operating system.
     * The Unicode representation of a character can occupy several bytes,
     * so the maximum number of characters that comprises a path and file name can vary.
     * The actual limitation is the number of bytes in the path and file components,
     * which might correspond to an equal number of characters.
     */
    private static final Integer MAX_TOPIC_LENGTH = 255;


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消息存储前要执行的校验方法，该方法在消息存储前执行一系列检查，确保Broker处于可写状态且消息格式合法
     * 包括检查存储服务是否关闭、是否为从节点、消息存储引擎是否可写、主题长度是否合法、消息体是否为空等，如果任何检查失败，则返回对应的错误结果，阻止消息写入
     */
    public static PutMessageResult checkBeforePutMessage(BrokerController brokerController, final MessageExt msg) {
        if (brokerController.getMessageStore().isShutdown()) {
            LOG.warn("message store has shutdown, so putMessage is forbidden");
            return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
        }
        if (!brokerController.getMessageStoreConfig().isDuplicationEnable() && BrokerRole.SLAVE == brokerController.getMessageStoreConfig().getBrokerRole()) {
            long value = PRINT_TIMES.getAndIncrement();
            if ((value % 50000) == 0) {
                LOG.warn("message store is in slave mode, so putMessage is forbidden ");
            }
            return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
        }
        if (!brokerController.getMessageStore().getRunningFlags().isWriteable()) {
            long value = PRINT_TIMES.getAndIncrement();
            if ((value % 50000) == 0) {
                LOG.warn("message store is not writeable, so putMessage is forbidden " + brokerController.getMessageStore().getRunningFlags().getFlagBits());
            }
            return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
        } else {
            PRINT_TIMES.set(0);
        }
        final byte[] topicData = msg.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
        boolean retryTopic = msg.getTopic() != null && msg.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX);
        if (!retryTopic && topicData.length > Byte.MAX_VALUE) {
            LOG.warn("putMessage message topic[{}] length too long {}, but it is not supported by broker",
                    msg.getTopic(), topicData.length);
            return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
        }
        if (topicData.length > MAX_TOPIC_LENGTH) {
            LOG.warn("putMessage message topic[{}] length too long {}, but it is not supported by broker",
                    msg.getTopic(), topicData.length);
            return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
        }
        if (msg.getBody() == null) {
            LOG.warn("putMessage message topic[{}], but message body is null", msg.getTopic());
            return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
        }
        if (brokerController.getMessageStore().isOSPageCacheBusy()) {
            return new PutMessageResult(PutMessageStatus.OS_PAGE_CACHE_BUSY, null);
        }
        return null;
    }

    // public static PutMessageResult checkInnerBatch(BrokerController brokerController, final MessageExt msg) {
    //     if (msg.getProperties().containsKey(MessageConst.PROPERTY_INNER_NUM)
    //             && !MessageSysFlag.check(msg.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG)) {
    //         LOG.warn("[BUG]The message had property {} but is not an inner batch", MessageConst.PROPERTY_INNER_NUM);
    //         return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
    //     }
    //
    //     if (MessageSysFlag.check(msg.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG)) {
    //         Optional<TopicConfig> topicConfig = Optional.ofNullable(brokerController.getTopicConfigManager().getTopicConfigTable().get(msg.getTopic()));
    //         if (!QueueTypeUtils.isBatchCq(topicConfig)) {
    //             LOG.error("[BUG]The message is an inner batch but cq type is not batch cq");
    //             return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
    //         }
    //     }
    //
    //     return null;
    // }
    //
    // public static PutMessageResult handleScheduleMessage(BrokerController brokerController,
    //                                                      final MessageExtBrokerInner msg) {
    //     final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
    //     if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
    //             || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
    //         if (!isRolledTimerMessage(msg)) {
    //             if (checkIfTimerMessage(msg)) {
    //                 if (!brokerController.getMessageStoreConfig().isTimerWheelEnable()) {
    //                     //wheel timer is not enabled, reject the message
    //                     return new PutMessageResult(PutMessageStatus.WHEEL_TIMER_NOT_ENABLE, null);
    //                 }
    //                 PutMessageResult transformRes = transformTimerMessage(brokerController, msg);
    //                 if (null != transformRes) {
    //                     return transformRes;
    //                 }
    //             }
    //         }
    //         // Delay Delivery
    //         if (msg.getDelayTimeLevel() > 0) {
    //             transformDelayLevelMessage(brokerController, msg);
    //         }
    //     }
    //     return null;
    // }
    //
    // private static boolean isRolledTimerMessage(MessageExtBrokerInner msg) {
    //     return TimerMessageStore.TIMER_TOPIC.equals(msg.getTopic());
    // }
    //
    // public static boolean checkIfTimerMessage(MessageExtBrokerInner msg) {
    //     if (msg.getDelayTimeLevel() > 0) {
    //         if (null != msg.getProperty(MessageConst.PROPERTY_TIMER_DELIVER_MS)) {
    //             MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_TIMER_DELIVER_MS);
    //         }
    //         if (null != msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC)) {
    //             MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_TIMER_DELAY_SEC);
    //         }
    //         if (null != msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS)) {
    //             MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_TIMER_DELAY_MS);
    //         }
    //         return false;
    //         //return this.defaultMessageStore.getMessageStoreConfig().isTimerInterceptDelayLevel();
    //     }
    //     //double check
    //     if (TimerMessageStore.TIMER_TOPIC.equals(msg.getTopic()) || null != msg.getProperty(MessageConst.PROPERTY_TIMER_OUT_MS)) {
    //         return false;
    //     }
    //     return null != msg.getProperty(MessageConst.PROPERTY_TIMER_DELIVER_MS) || null != msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS) || null != msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC);
    // }
    //
    // private static PutMessageResult transformTimerMessage(BrokerController brokerController,
    //                                                       MessageExtBrokerInner msg) {
    //     //do transform
    //     int delayLevel = msg.getDelayTimeLevel();
    //     long deliverMs;
    //     try {
    //         if (msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC) != null) {
    //             deliverMs = System.currentTimeMillis() + Long.parseLong(msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC)) * 1000;
    //         } else if (msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS) != null) {
    //             deliverMs = System.currentTimeMillis() + Long.parseLong(msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS));
    //         } else {
    //             deliverMs = Long.parseLong(msg.getProperty(MessageConst.PROPERTY_TIMER_DELIVER_MS));
    //         }
    //     } catch (Exception e) {
    //         return new PutMessageResult(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, null);
    //     }
    //     if (deliverMs > System.currentTimeMillis()) {
    //         if (delayLevel <= 0 && deliverMs - System.currentTimeMillis() > brokerController.getMessageStoreConfig().getTimerMaxDelaySec() * 1000L) {
    //             return new PutMessageResult(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, null);
    //         }
    //
    //         int timerPrecisionMs = brokerController.getMessageStoreConfig().getTimerPrecisionMs();
    //         if (deliverMs % timerPrecisionMs == 0) {
    //             deliverMs -= timerPrecisionMs;
    //         } else {
    //             deliverMs = deliverMs / timerPrecisionMs * timerPrecisionMs;
    //         }
    //
    //         if (brokerController.getTimerMessageStore().isReject(deliverMs)) {
    //             return new PutMessageResult(PutMessageStatus.WHEEL_TIMER_FLOW_CONTROL, null);
    //         }
    //         MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TIMER_OUT_MS, deliverMs + "");
    //         MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
    //         MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
    //         msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
    //         msg.setTopic(TimerMessageStore.TIMER_TOPIC);
    //         msg.setQueueId(0);
    //     } else if (null != msg.getProperty(MessageConst.PROPERTY_TIMER_DEL_UNIQKEY)) {
    //         return new PutMessageResult(PutMessageStatus.WHEEL_TIMER_MSG_ILLEGAL, null);
    //     }
    //     return null;
    // }
    //
    // public static void transformDelayLevelMessage(BrokerController brokerController, MessageExtBrokerInner msg) {
    //
    //     if (msg.getDelayTimeLevel() > brokerController.getScheduleMessageService().getMaxDelayLevel()) {
    //         msg.setDelayTimeLevel(brokerController.getScheduleMessageService().getMaxDelayLevel());
    //     }
    //
    //     // Backup real topic, queueId
    //     MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
    //     MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
    //     msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
    //
    //     msg.setTopic(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC);
    //     msg.setQueueId(ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel()));
    // }

    public static boolean sendMessageBack(BrokerController brokerController, List<MessageExt> msgList,
                                          String brokerName, String brokerAddr) {
        try {
            Iterator<MessageExt> it = msgList.iterator();
            while (it.hasNext()) {
                MessageExt msg = it.next();
                msg.setWaitStoreMsgOK(false);
                //brokerController.getBrokerOuterAPI().sendMessageToSpecificBroker(brokerAddr, brokerName, msg, "InnerSendMessageBackGroup", 3000);
                it.remove();
            }
        } catch (Exception e) {
            LOG.error("send message back to broker {} addr {} failed", brokerName, brokerAddr, e);
            return false;
        }
        return true;
    }
}

